package com.zxy.zxyrpc.server.tcp;

import cn.hutool.core.util.IdUtil;
import com.zxy.zxyrpc.exception.ConnectException;
import com.zxy.zxyrpc.RpcApplication;
import com.zxy.zxyrpc.exception.RpcException;
import com.zxy.zxyrpc.model.ErrorCode;
import com.zxy.zxyrpc.model.RpcRequest;
import com.zxy.zxyrpc.model.RpcResponse;
import com.zxy.zxyrpc.model.ServiceMetaInfo;
import com.zxy.zxyrpc.protocol.*;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetClient;
import io.vertx.core.net.NetSocket;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class VertxTcpClient {
    public static void main(String[] args) {
        new VertxTcpClient().start();
    }

    public static RpcResponse doRequest(RpcRequest req, ServiceMetaInfo serviceMetaInfo) throws ExecutionException, InterruptedException {
        // 发送 TCP 请求
        Vertx vertx = Vertx.vertx();
        NetClient netClient = vertx.createNetClient();
        // 异步变同步处理
        CompletableFuture<RpcResponse> responseFuture = new CompletableFuture<>();
        try {
            netClient.connect(serviceMetaInfo.getServicePort(), serviceMetaInfo.getServiceHost(),
                    result -> {
                        if (!result.succeeded()) {
                            System.err.println("连接 TCP 服务失败");
                            // 向外抛出异常，便于重试机制捕捉（retry的重试机制依赖于异常捕捉）
                            responseFuture.completeExceptionally(new RuntimeException("连接 TCP 服务失败"));
                            return;
                        }
                        // 构建协议信息
                        ProtocolMessage.Header header = new ProtocolMessage.Header();
                        header.setMagic(ProtocolConstant.PROTOCOL_MAGIC);
                        header.setVersion(ProtocolConstant.PROTOCOL_VERSION);
                        header.setSerializer((byte) ProtocolMessageSerializerEnum.getEnumByValue(RpcApplication.getRpcConfig().getSerializer()).getKey());
                        header.setType((byte) ProtocolMessageTypeEnum.REQ.getKey());
                        header.setRequestId(IdUtil.getSnowflake(1).nextId());

                        ProtocolMessage<RpcRequest> protocolMessage = new ProtocolMessage<>();
                        protocolMessage.setHeader(header);
                        protocolMessage.setBody(req);

                        // 编码请求并发送
                        NetSocket socket = result.result();
                        try {
                            Buffer encode = ProtocolMessageEncoder.encode(protocolMessage);
                            socket.write(encode);
                        } catch (IOException e) {
                            throw new RpcException(ErrorCode.OPERATION_ERROR, "协议消息编码错误");
                        }

                        // 增强 handle 对 buffer 进行处理
                        TcpBufferHandleWrapper bufferHandleWrapper = new TcpBufferHandleWrapper(buffer -> {
                            try {
                                ProtocolMessage<RpcResponse> decode = (ProtocolMessage<RpcResponse>) ProtocolMessageDecoder.decode(buffer);
                                responseFuture.complete(decode.getBody());
                            } catch (IOException e) {
                                throw new RpcException(ErrorCode.OPERATION_ERROR, "协议消息解码错误");
                            }
                        });
                        // 调用请求处理器
                        socket.handler(bufferHandleWrapper);
                    });
        } catch (Exception e) {
            throw new ConnectException("连接失效: " + serviceMetaInfo.getServiceHost() + ":" + serviceMetaInfo.getServicePort());
        }
        RpcResponse rpcResponse = responseFuture.get();
        // 关闭 TCP连接
        netClient.close();
        return rpcResponse;
    }

    /**
     * 测试半包，粘包
     */
    public void start() {
        // 创建 vertx 实例
        Vertx vertx = Vertx.vertx();
        vertx.createNetClient().connect(8888, "localhost", result -> {
            if (result.succeeded()) {
                System.out.println("连接 TCP 服务成功!");
                NetSocket socket = result.result();
                for (int i = 0; i < 1000; i++) {
                    Buffer buffer = Buffer.buffer();
                    String str = "Hello, server!Hello, server!Hello, server!";
                    buffer.appendInt(0);
                    buffer.appendInt(str.getBytes().length);
                    buffer.appendBytes(str.getBytes());
                    socket.write(buffer);
                }
                // 发送数据
                // socket.write("Hello,Server");
                // 接收响应
                socket.handler(buffer -> System.out.println("服务端响应: " + buffer.toString()));

            } else
                System.out.println("连接服务端失败");
        });
    }
}
